package main
//消费者
import (
	"fmt"
	kafka "github.com/Shopify/sarama"
	"log"
)

func main() {
	cousumer,err :=  kafka.NewConsumer([]string{"127.0.0.1:9092"},nil)
	if err!=nil {
		log.Fatal("fail to start consumer",err)
		return
	}
	// 根据topic取到所有的分区
	partitionList,err :=  cousumer.Partitions("web_log")
	if err !=nil {
		log.Fatal("fail to get list of partition:err%v\n", err)
		return
	}
	fmt.Println(partitionList)
	// 遍历所有的分区
	for partition := range partitionList{
		pc, err :=cousumer.ConsumePartition("web_log", int32(partition),kafka.OffsetNewest)
		if err !=nil {
			log.Fatal("failed to start consumer for partition %d,err:%v\n", partition, err)
			return
		}
		defer pc.AsyncClose()
		// 异步从每个分区消费信息
		go func(kafka.PartitionConsumer) {
			for msg := range pc.Messages() {
				fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v", msg.Partition, msg.Offset, msg.Key, msg.Value)
			}
		}(pc)
	}
}
